// псевдокод

while (1) {
    if (is_ready(alice_fd)) {
        work_with(alice_fd);
    } else if (is_ready(bob_fd)) {
        work_with(bob_fd);
    } else {
        sleep();
    }
}


%%writefile example1.c

#include <fcntl.h>
#include <stdio.h>
#include <unistd.h>

int main() {
    char buffer[123];
    fcntl(0, F_SETFL, O_NONBLOCK);  // Вся магия тут!)
    int len = read(0, buffer, sizeof(buffer));
    printf("%d\n", len);
    return 0;
}

Overwriting example1.c


!gcc example1.c -o example1
!./example1

-1

[danila@archlinux ~/caos/14-async & multithreading]
$ ./example1
-1

while (1) {
    int ready_fd = get_ready_client();

    work_with(ready_fd);
}


%%writefile example2.c

#include <stdio.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <signal.h>
#include <sched.h>

int exit_pipe[2];


void handler(int sig) {
    char msg[] = "123";
    write(exit_pipe[1], msg, sizeof(msg));
}


int main() {
    pipe(exit_pipe);

    dprintf(2, "signal\n");
    struct sigaction sa = {
        .sa_handler = handler,
        .sa_flags = SA_RESTART
    };
    
    sigaction(SIGINT, &sa, NULL);
    
    int epoll_fd = epoll_create1(0);  // no flags
    
    fcntl(exit_pipe[0], F_SETFL, O_NONBLOCK);
    fcntl(0, F_SETFL, O_NONBLOCK);
    
    int fifo_fd = open("test_fifo", O_RDONLY);
    fcntl(fifo_fd, F_SETFL, O_NONBLOCK);
    
    dprintf(2, "epoll_ctl...\n");
    int all_fds[] = {exit_pipe[0], 0, fifo_fd};
    for (int i = 0; i < 3; ++i) {
        struct epoll_event event = {
            .events = EPOLLIN,
            .data.fd = all_fds[i]
        };
        
        // Будем ожидать готовности на чтение
        epoll_ctl(epoll_fd, EPOLL_CTL_ADD, all_fds[i], &event);
    }
    
    dprintf(2, "Main loop\n");
    while (1) {
        struct epoll_event event;
        int N = epoll_wait(epoll_fd, &event, /*maxevents=*/ 1, /*timeout=*/ 1000);
        
        if (N <= 0) {
            sched_yield();
            continue;
        }
        
        int fd = event.data.fd;
        
        if (fd == exit_pipe[0]) {
            break;
        }
        
        char buffer[1024];
        size_t len = read(fd, buffer, sizeof(buffer));
        write(1, buffer, len);
    }
    
    dprintf(2, "Closing all\n");
    
    for (int i = 0; i < 3; ++i) {
        close(all_fds[i]);
    }
    
    close(epoll_fd);
    
    dprintf(2, "Exiting...\n");
    
    return 0;
}

Overwriting example2.c

# term0
[danila@archlinux ~/caos/14-async & multithreading]
$ ./example2
signal
epoll_ctl...
Main loop
1234
1234
abc
^CClosing all
Exiting...
[danila@archlinux ~/caos/14-async & multithreading]
$

#include <pthread.h>

void *worker(void *args) {
    /// ...
}

int main() {
    pthread_t thread;

    int my_int_arg;

    pthread_create(
        &thread,
        NULL /*init and destroy thread attributes*/,
        worker,
        (void*)(&my_int_arg)
    );

    // ...

    pthread_join(&thread, NULL /*retval*/);
}

// не атомарно
item_t *new_node = malloc(...);

do {
    // тоже не атомарно
    new_node->next = head;

    // опять не атомарно, но где-то там есть что-то атомарное
} while (!atomic_compare_exchange_weak(&head, &new_node->next, (_Atomic item_t*)new_node));


%%writefile example3.c

// without mutex

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

int global_int;

void *worker(void *arg) {
    int *gi = (int*)arg;
    
    for (int i = 0; i < 1000; ++i) {
        int old = *gi;
        int new = old + 1;
        *gi = new;
    }
}

int main() {
    global_int = 0;
    pthread_t threads[2];
    
    pthread_create(threads + 0, NULL, worker, (void*)(&global_int));
    pthread_create(threads + 1, NULL, worker, (void*)(&global_int));
    
    pthread_join(threads[0], NULL);
    pthread_join(threads[1], NULL);
    
    printf("%d\n", global_int);
    
    return 0;
}

Overwriting example3.c


!gcc example3.c -o example3
!./example3

1980


%%writefile example4.c

// with mutex

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

pthread_mutex_t mutex;
int global_int;

void *worker(void *arg) {
    int *gi = (int*)arg;
    
    for (int i = 0; i < 1000; ++i) {
        pthread_mutex_lock(&mutex);
        
        int old = *gi;
        int new = old + 1;
        *gi = new;
        
        pthread_mutex_unlock(&mutex);
    }
}

int main() {
    global_int = 0;
    pthread_t threads[2];
    
    pthread_create(threads + 0, NULL, worker, (void*)(&global_int));
    pthread_create(threads + 1, NULL, worker, (void*)(&global_int));
    
    pthread_join(threads[0], NULL);
    pthread_join(threads[1], NULL);
    
    printf("%d\n", global_int);
    
    return 0;
}

Writing example4.c


!gcc example4.c -o example4
!./example4

2000


%%writefile example5.c

#include <stdio.h>
#include <stdlib.h>

#include <pthread.h>
#include <unistd.h>

typedef struct {
  pthread_mutex_t *mutex;
  pthread_cond_t *condvar;

  int64_t *next_out;
} gen_args_t;


void *do_work(void *args) {
  gen_args_t *params = (gen_args_t*)args;

  for (int64_t i=0; ; ++i) {
    pthread_mutex_lock(params->mutex);
    *params->next_out = i;

    pthread_cond_signal(params->condvar);

    while (*params->next_out != -1)
        pthread_cond_wait(params->condvar, params->mutex);

    pthread_mutex_unlock(params->mutex);
  }
}

int main(int argc, char **argv) {
  pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  pthread_cond_t condvar = PTHREAD_COND_INITIALIZER;
  int64_t next_out;
  gen_args_t thread_args = {
      .mutex = &mutex,
      .condvar = &condvar,
      .next_out = &next_out,
  };

  pthread_t worker;
  pthread_create(&worker, NULL, do_work, &thread_args);

  next_out = -1;
  const int N = 10;
  for (int32_t count = 0; count < N; ++count) {
    pthread_mutex_lock(&mutex);
      
    while (next_out == -1)
      pthread_cond_wait(&condvar, &mutex);

    int64_t cur_out = next_out;

    next_out = -1;
    pthread_mutex_unlock(&mutex);
  
    if (count != N - 1)
      pthread_cond_signal(&condvar);

    printf("%lld\n", cur_out);
  }

  pthread_cancel(worker);

  pthread_cond_destroy(&condvar);
  pthread_mutex_destroy(&mutex);
  return 0;
}

Overwriting example5.c


!gcc example5.c -o example5
!./example5

CAOS-14 async & multithreading¶

Асинхронность¶

Intro¶

epoll¶

Нулевка¶

Многопоточность¶

Intro¶

Потоки¶

(Не)атомарность¶

Mutex¶

Condvar (conditional variable)¶

Lock-free¶